package mqtt;

import com.szh.common.constant.MQTTConstants;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

/**
 * mqtt 发布消息
 */
@Slf4j
public class PublishMsg {

    private static MqttClient client = null;

    static {
        try {
            MemoryPersistence persistence = new MemoryPersistence();
            client = new MqttClient(MQTTConstants.MQTT_BROKER_URL, MQTTConstants.PUBLISH_MQTT_CLIENT_ID, persistence);
            // MQTT 连接选项
            MqttConnectOptions connOpts = new MqttConnectOptions();
            connOpts.setUserName(MQTTConstants.mqtt_username);
            connOpts.setPassword(MQTTConstants.mqtt_pwd.toCharArray());
            // 保留会话
            connOpts.setCleanSession(false);
            // 建立连接
            log.error("Connecting to broker: " + MQTTConstants.MQTT_BROKER_URL);
            client.connect(connOpts);
            log.error("Connected");
            log.error("------------------");
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    
    public static void main(String[] args) throws MqttException, InterruptedException {
        // 消息发布所需参数
        for (int i = 0; i < 1000; i++) {
            String msg ="Hello World-"+i;
            MqttMessage message = new MqttMessage(msg.getBytes());
            message.setQos(2);
            client.publish(MQTTConstants.mqtt_publish_topic, message);
            log.error("Message published- 成功");
            Thread.sleep(4000);
        }

        client.disconnect();
        log.error("Disconnected");
        client.close();

    }}

